import io.reactivex.Observable
import io.reactivex.plugins.RxJavaPlugins
import io.reactivex.schedulers.Schedulers
import io.reactivex.subjects.PublishSubject
import org.junit.Before
import org.junit.Test

/**
 * 消费者比生产者慢时
 *
 * 消费者怎么同时处理多个事件，避免阻塞
 */
class MultiSubscribe {

    private fun printWithTime(a: Any) {
        println("${System.currentTimeMillis()} $a")
    }


    @Test //结论：发一个收一个；
    fun testSingle() {
        val publishSubject = PublishSubject.create<Int>()
        publishSubject
            .doOnNext {
                printWithTime("Receive $it")
                Thread.sleep(10)
                if (it == 4) {
                    throw RuntimeException("mock failed")
                }
            }
            .retry()
            .subscribeOn(Schedulers.trampoline())
            .subscribe {  printWithTime("End $it")}

        //单线程测试
        Observable.just(1, 2, 3, 4, 5)
            .subscribe(publishSubject::onNext)

        Thread.sleep(2000)
    }

    @Test //结论：和单线程一样发一个收一个
    fun testObserveOn() {
        val publishSubject = PublishSubject.create<Int>()
        publishSubject
//            .observeOn(Schedulers.newThread())//
            .observeOn(Schedulers.io())//和单线程一样发一个收一个
            .doOnNext {
                printWithTime("Receive $it")
                Thread.sleep(10)
                if (it == 4) {
                    throw RuntimeException("mock failed")
                }
            }
            .retry()
            .subscribeOn(Schedulers.trampoline())
            .subscribe {  printWithTime("End $it")}

        //单线程测试
        Observable.just(1, 2, 3, 4, 5)
            .subscribeOn(Schedulers.trampoline())
            .subscribe(publishSubject::onNext)


        Thread.sleep(2000)
    }


    @Test //结论：通过flat, 使用新线程可以达到并发处理的效果。并发所以不一定谁会先执行
    fun testFlatMap() {
        val publishSubject = PublishSubject.create<Int>()
        publishSubject
            .flatMap {
               Observable.fromCallable {
                   printWithTime("Receive $it")
                   Thread.sleep(10)
                   if (it == 4) {
                       throw RuntimeException("mock failed")
                   }
                   it
               }
                   .onErrorReturn { -1  } //这边要处理error，然后可选在下一步过滤
                   .filter { filterIt -> filterIt > 0 }
                   .subscribeOn(Schedulers.newThread())

            }
            .retry()
            .subscribeOn(Schedulers.trampoline())
            .subscribe {  printWithTime("End $it")}

        //单线程测试
        Observable.just(1, 2, 3, 4, 5)
            .subscribeOn(Schedulers.trampoline())
            .subscribe(publishSubject::onNext)


        Thread.sleep(2000)
    }

    @Before
    fun setUp() {
        RxJavaPlugins.setErrorHandler {
            println(it)
        }
    }
}